package net.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.MessageToMessageCodec;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 1. Netty抽象出两组线程池，parentGroup专门负责接收客户端的连接，childGroup专门负责网络的读写
 * <p>
 * 2. parentGroup和childGroup类型的本质都是NioEventLoopGroup类型。
 * <p>
 * 3. NioEventLoopGroup相当于一个线程管理器（类似于ExecutorService），它下面维护很多个NioEventLoop线程。
 * <p>
 * * 3.1 在初始化这两个Group线程组时，默认会在每个Group中生成CPU*2个NioEventLoop线程
 * * 3.2 当n个连接来了，Group默认会按照连接请求的顺序分别将这些连接分给各个NioEventLoop去处理。
 * * 3.3 同时Group还负责管理EventLoop的生命周期。
 * <p>
 * 4. NioEventLoop表示一个不断循环的执行处理任务的线程
 * * 4.1 它维护了一个线程和任务队列。
 * * 4.2 每个NioEventLoop都包含一个Selector，用于监听绑定在它上面的socket通讯。
 * * 4.3 每个NioEventLoop相当于Selector，负责处理多个Channel上的事件
 * * 4.4 每增加一个请求连接，NioEventLoopGroup就将这个请求依次分发给它下面的NioEventLoop处理。
 * <p>
 * 5. 每个parent NioEventLoop循环执行的步骤有3步：
 * * 5.1 轮询accept事件
 * * 5.2 处理accept事件，与client建立连接，生成NioSocketChannel，并将其注册到某个child NioEventLoop的selector上。
 * * 5.3 处理任务队列到任务，即runAllTasks
 * <p>
 * 6. 每个child NioEventLoop循环执行的步骤：
 * * 6.1 轮询read，write事件
 * * 6.2 处理I/O事件，即read，write事件，在对应的NioSocketChannel中进行处理
 * * 6.3 处理任务队列的任务，即runAllTasks
 * <p>
 * 7. 每个 child NioEventLoop处理业务时，会使用pipeline（管道）
 * * 7.1 pipeline中维护了一个ChannelHandlerContext链表
 * * 7.2 ChannelHandlerContext则保存了Channel相关的所有上下文信息，同时关联一个ChannelHandler（处理器）对象
 * * 7.3 Channel和pipeline一一对应，ChannelHandler和ChannelHandlerContext一一对应
 * <p>
 * 8. 任务队列
 * 任务队列由NioEventLoop维护并不断执行。当我们就收到请求之后，在当前channel对应的pipeline中的各个Handler里面进行业务处理和请求过滤。
 * 当某些业务需要耗费大量时间的时候，我们可以将任务提交到由NioEventLoop维护的taskQueue或scheduleTaskQueue中，让当前的NioEventLoop线程在空闲时间去执行这些任务
 * <p>
 * 9. Future
 * * 9.1 表示异步的执行结果, 可以通过它提供的方法来检测执行是否完成，比如检索计算等等
 * * 9.2 ChannelFuture 是一个接口：public interface ChannelFuture extends Future<Void>。我们可以添加监听器，当监听的事件发生时，就会通知到监听器
 *
 * @author 王霄
 * @date 2021/8/28 11:27
 */
public class NettyServer {

    public static void main(String[] args) {

        //创建parentGroup(只处理连接请求),无限循环
        EventLoopGroup parentGroup = new NioEventLoopGroup();

        //创建childGroup(与客户端业务处理),无限循环
        EventLoopGroup childGroup = new NioEventLoopGroup();

        try {
            //创建服务器端的启动对象,配置参数
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(parentGroup, childGroup)//设置两个线程组
                    .channel(NioServerSocketChannel.class)//使用NioSocket作为服务器的通道实现
                    .option(ChannelOption.SO_BACKLOG, 128)//设置线程队列连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() {//给childGroup设置对应的处理器
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new NettyServerHandle());
                        }
                    });
            //绑定一个端口并且同步,生成一个ChannelFuture对象(启动了服务器)
            ChannelFuture channelFuture = bootstrap
                    .bind(7777).sync()
                    .addListener(future -> {
                        if (future.isSuccess()) {
                            System.out.println("端口7777绑定成功");
                        } else {
                            System.out.println("端口7777绑定失败");
                        }
                    });

            //对关闭通道进行监听
            Channel channel = channelFuture.channel();
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }


    public static class NettyServerHandle extends MessageToMessageCodec<ByteBuf,String> {

        @Override
        protected void encode(ChannelHandlerContext ctx, String s, List<Object> list) {

        }

        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> list) {
            //比如有一个耗时的操作
            ctx.channel().eventLoop().execute(new Task1());
            ctx.channel().eventLoop().schedule(new Task2(), 60, TimeUnit.SECONDS);

            System.out.println("ctx = " + ctx);
            System.out.println("客户端发送的消息是：" + buf.toString(StandardCharsets.UTF_8));
            System.out.println("客户端地址：" + ctx.channel().remoteAddress());
        }

        //数据读取完毕
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            //将数据写入缓冲并刷新
            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端", StandardCharsets.UTF_8));
        }

        //监听异常
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

    public static class Task1 implements Runnable {
        @Override
        public void run() {
            try {
                Thread.sleep(10 * 1000);
                System.out.println("任务1处理完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static class Task2 implements Runnable {
        @Override
        public void run() {
            try {
                Thread.sleep(10 * 1000);
                System.out.println("任务2处理完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
